package  com.qyer.log.job.sum;
import java.io.IOException;
import java.net.URISyntaxException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import com.qyer.util.log.util.MyDateUtil;

public class UserTracksSummaryWeek {

	
	public static class M1 extends Mapper<LongWritable, Text, Text, Text> {
		private Text oKey = new Text();
		private Text oVal = new Text();
		final String FS = "\t";

		public void map(LongWritable key, Text value, Context context) throws IOException,InterruptedException {
			String []fields = value.toString().split(FS); 
			
			String vkey = fields[4]+FS+fields[5]+FS+fields[6]+FS+fields[7]; 
			String vval = fields[8]; 
			oKey.set(vkey);
			oVal.set(vval);
			context.write(oKey, oVal);
		}
	}

	public static class C1 extends Reducer<Text, Text, Text, Text> {
		Text oKey = new Text();
		Text oVal = new Text();
		String DATE;
		String LEN; 
		final String FS = "\t";
		
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,
				InterruptedException {
			
			Set<String> cookieIdSet = new HashSet<String>();
			
			for (Text value : values){
				String uvStr  = value.toString(); 	
				
				uvStr=StringUtils.replace(uvStr , "{", "");
				uvStr=StringUtils.replace(uvStr , "}", "");
				
				if(uvStr.indexOf(",")>-1){
					for(String uvS : uvStr.split(",")){
						cookieIdSet.add(uvS); 
					}
				}else{
					cookieIdSet.add(uvStr); 
				}
  			}
			
			oKey.set(key);
			oVal.set(cookieIdSet.toString());
			context.write(oKey, oVal);
		}
	}
	public static class R1 extends Reducer<Text, Text, Text, Text> {
		Text oKey = new Text();
		Text oVal = new Text();
		String DATE;
		String LEN; 
		final String FS = "\t";
		
		public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,
				InterruptedException {
			
			Set<String> cookieIdSet = new HashSet<String>();
  			long uv = 0l;
			
			if (DATE == null) {
				DATE = context.getConfiguration().get("DATE");
			}
			
			if (LEN == null) {
				LEN = context.getConfiguration().get("LEN");
			}

			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
  			Long curTimes=0L;
			try {
				String d =sdf.format(new Date());
				curTimes = sdf.parse(d).getTime();
			} catch (ParseException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			curTimes=curTimes/1000;
			
			
			for (Text value : values){
				String uvStr  = value.toString(); 	
				
				uvStr=StringUtils.replace(uvStr , "[", "");
				uvStr=StringUtils.replace(uvStr , "]", "");
				
				if(uvStr.indexOf(",")>-1){
					for(String uvS : uvStr.split(",")){
						cookieIdSet.add(uvS); 
					}
				}else{
					cookieIdSet.add(uvStr); 
				}
  			}
			
			uv=cookieIdSet.size();
			
			oKey.set(curTimes+FS +DATE+FS+uv+FS+LEN);
			if(LEN.equals("7") || LEN.equals("30")){
				oVal.set(key);
			}else{
				oVal.set(key+FS+cookieIdSet.toString());
			}
			
			context.write(oKey, oVal);
		}
	}

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException,
			URISyntaxException, ParseException {

		Configuration conf = new Configuration();
		conf.set("mapred.job.queue.name", "regular"); // default,regular,realtime
//		conf.set("mapreduce.map.memory.mb" ,"8080");
		conf.set("mapreduce.reduce.memory.mb" ,"8080");
//		conf.set("mapreduce.map.java.opts", "-Xmx8080m");
		conf.set("mapreduce.reduce.java.opts", "-Xmx8080m");
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		int exitCode = 127;
		
		if (otherArgs.length != 2) {
			System.out.println("************************************************************");
			System.out.println("************************************************************");
			System.out.println("Usage: please input 1 params, for example: file.jar args[0]");
			System.out.println("args[0] is 20130106");
			System.out.println("args[1] is 7 or 10 or  30 ");
			System.out.println("*********************************	***************************");
			System.out.println("************************************************************");
			System.exit(exitCode);
		}
		
		String startDate = otherArgs[0].split(",")[0];
		String endDate = otherArgs[0].split(",").length == 2 ? otherArgs[0].split(",")[1] : startDate;
		int timeCycle = Integer.parseInt(otherArgs[1]); 	
		
//		int days = Integer.parseInt(otherArgs[2]); 	
		
		endDate=MyDateUtil.getNDaysAgo(startDate , -(timeCycle-1));
		
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		Long startDateLong =sdf.parse(startDate).getTime();
		startDateLong=startDateLong/1000;
		conf.set("DATE", startDateLong.toString());
		if(timeCycle==7){
			conf.set("LEN", "7");
		}else if (timeCycle==10){
			conf.set("LEN", "10");
		}else{
			conf.set("LEN", "30");
		}
		
		String dayInPath = "/flume/summary/UserTracksSummary/day";
		String weekOutPath = "/flume/summary/UserTracksSummary/week";
		String tendaysOutPath = "/flume/summary/UserTracksSummary/tendays";
		String mouthOutPath = "/flume/summary/UserTracksSummary/mouth";
		

		Job job = new Job(conf, "UserTracksSummaryWeek");
		job.setInputFormatClass(TextInputFormat.class);
		job.setJarByClass(UserTracksSummaryWeek.class);
		job.setMapperClass(M1.class);
		job.setCombinerClass(C1.class);
		job.setReducerClass(R1.class);
		job.setNumReduceTasks(1);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		if(timeCycle==7){
			for (String runDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
		        System.out.println(  " add input path :"+runDate );
		        FileInputFormat.addInputPath(job, new Path(dayInPath + "/"+runDate+"/*"));
			}
			
			String outPath = weekOutPath + "/" + startDate;
			FileSystem.get(conf).delete(new Path(outPath), true);
			FileOutputFormat.setOutputPath(job, new Path(outPath));
		}else if (timeCycle==10||timeCycle==30 ||timeCycle==31||timeCycle==28){
			for (String runDate : MyDateUtil.getDateList(startDate, endDate, 0)) {
		        System.out.println(  " add input path :"+runDate );
		        FileInputFormat.addInputPath(job, new Path(dayInPath + "/"+runDate+"/*"));
			}
			
			if (timeCycle==10){
				String outPath = tendaysOutPath + "/" + startDate;
				FileSystem.get(conf).delete(new Path(outPath), true);
				FileOutputFormat.setOutputPath(job, new Path(outPath));
			}else{
				String outPath = mouthOutPath + "/" + startDate;
				FileSystem.get(conf).delete(new Path(outPath), true);
				FileOutputFormat.setOutputPath(job, new Path(outPath));
			}
		}else if (timeCycle==32){
			
			String runDate = startDate.substring(0 , 6 ); 
			System.out.println(  " add input path :"+runDate );
		    FileInputFormat.addInputPath(job, new Path(tendaysOutPath + "/"+runDate+"*/*"));
			
//			for (String runDate : MyDateUtil.getDateList(startDate, endDate, 10)) {
//		       
//			}
			
			String outPath = mouthOutPath + "/" + startDate;
			FileSystem.get(conf).delete(new Path(outPath), true);
			FileOutputFormat.setOutputPath(job, new Path(outPath));
		}
		
		
		exitCode = job.waitForCompletion(true) ? 0 : 1;
		System.out.println("--------------------------------------------END1");

		System.exit(exitCode);

	}

}
